﻿using System;

using Adaptive.Agrona;
using Adaptive.Agrona.Concurrent;
using Adaptive.Agrona.Concurrent.RingBuffer;

using Microsoft.Extensions.Logging;

using Unity.Connections;
using Unity.Commons.Client.Generator;
using Unity.Logger;

namespace Unity.Commons.Client.Buffer
{
    public sealed class UnityRingBuffer : IUnityBuffer, IAgent
    {
        private static readonly ILogger Logger = UnityLogger.ConsoleLoggerFactory.CreateLogger(nameof(UnityRingBuffer));

        private const int MetadataTypeId = 1;
        private const int BatchingSize = 128;

        private readonly IRingBuffer _ringBuffer;

        private AgentRunner _agentRunner;
        private static IUnityProducer Producer;
        private static IUnityBuffer BackupBuffer;

        private static readonly AtomicLong Published = new AtomicLong();
        private static readonly AtomicLong Dropped = new AtomicLong();

        private static ClientMode Mode = ClientMode.Offline;

        private static volatile bool IsClosing;
        private static volatile bool IsFlushWaiting;
        private static volatile bool IsFailedWhileClosing;

        private static void SetIsClosing(bool value)
        {
            IsClosing = value;
        }

        private static void SetIsFlushWaiting(bool value)
        {
            IsFlushWaiting = value;
        }

        private static void SetIsFailedWhileClosing(bool value)
        {
            IsFailedWhileClosing = value;
        }
        
        public UnityRingBuffer(IUnityProducer producer, IUnityBuffer backupBuffer, IAtomicBuffer byteBuffer)
        {
            SetProducer(producer);
            SetBackupBuffer(backupBuffer);
            try
            {
                _ringBuffer = new ManyToOneRingBuffer(byteBuffer);
            }
            catch(Exception exception) 
            {
                Logger.LogError(exception, "Error initialize ring buffer in UnityRingBuffer");
            }
            
        }

        private static void SetProducer(IUnityProducer producer)
        {
            Producer = producer;
        }

        private static void SetBackupBuffer(IUnityBuffer buffer)
        {
            BackupBuffer = buffer;
        }

        public IAtomicBuffer GetBuffer()
        {
            return _ringBuffer.Buffer();
        }

        public IUnityProducer GetProducer()
        {
            return Producer;
        }

        public ClientMode GetMode()
        {
            return Mode;
        }

        int IUnityBuffer.Claim(int length)
        {
            return _ringBuffer.TryClaim(MetadataTypeId, length);
        }

        void IUnityBuffer.Commit(int index)
        {
            _ringBuffer.Commit(index);
        }

        void IUnityBuffer.Abort(int index)
        {
            _ringBuffer.Abort(index);
        }

        public static int GetRecordLength()
        {
            return BitUtil.Align(UnityGenerator.MessageLength + RecordDescriptor.HeaderLength,
                RecordDescriptor.Alignment);
        }

        public long GetBufferCapacity()
        {
            return _ringBuffer.Capacity() / GetRecordLength();
        }

        public long GetBufferSize()
        {
            return _ringBuffer.Size() / GetRecordLength();
        }

        public long GetPublished()
        {
            return Published.Get();
        }

        public long GetDropped()
        {
            return Dropped.Get();
        }

        private readonly ErrorHandler _errorHandler = (exception) => 
        {
            Logger.LogError(exception, "Unexpected error during processing ring buffer");
        };

        void IUnityBuffer.Start()
        {
            _agentRunner = new AgentRunner(new BusySpinIdleStrategy(), _errorHandler, null, this);
            AgentRunner.StartOnThread(_agentRunner);
        }

        public void WaitFor()
        {
            while (_ringBuffer.Size() > 0)
            {
                // waiting done operation in resource thread
            }
        }

        public void Flush()
        {
            SetIsFlushWaiting(true);
        }

        public void Close()
        {
            SetIsClosing(true);
            while (_ringBuffer.Size() > 0 && !IsFailedWhileClosing)
            {
                // waiting done operation in resource thread
            }
            if (_ringBuffer.Size() > 0)
            {
                Logger.LogWarning("Buffer will be closed with {} unread messages", _ringBuffer.Size() / GetRecordLength());
            }
            if(_agentRunner != null)
            {
                _agentRunner.Dispose();
            }
        }
        public void Close(long timeout)
        {
            SetIsClosing(true);
            long endAt = UnityTimestamp.ClockTimestamp() + timeout;
            while (true)
            {
                if (_ringBuffer.Size() <= 0 || IsFailedWhileClosing || UnityTimestamp.ClockTimestamp() >= endAt) break;
            }
            if (_ringBuffer.Size() > 0)
            {
                Logger.LogWarning("Buffer will be closed with {} unread messages", _ringBuffer.Size() / GetRecordLength());
            }
            _agentRunner.Dispose();

        }

        void IAgent.OnStart()
        {
            // Method intentionally left empty.
        }

        void IAgent.OnClose()
        {
            // Method intentionally left empty.
        }

        private readonly MessageHandler _ringBufferMessageHandler =
            (msgTypeId, buffer, index, length) =>
            {
                if (Producer.Produce(buffer, index, length))
                {
                    Published.IncrementAndGet();
                    Mode = ClientMode.Streaming;
                    Logger.LogTrace("Message Handler done processing with status: COMMIT");
                }
                else if (BackupBuffer != null)
                {
                    int claimedPosition = BackupBuffer.Claim(UnityGenerator.MessageLength);
                    if (claimedPosition > 0)
                    {
                        try
                        {
                            BackupBuffer.GetBuffer().PutBytes(claimedPosition, buffer, index, length);
                            BackupBuffer.Commit(claimedPosition);
                        }
                        catch (Exception exception)
                        {
                            BackupBuffer.Abort(claimedPosition);
                            Logger.LogError(exception, "Unexpected error during buffering, message will be aborted");
                        }
                    }
                    else
                    {
                        Dropped.IncrementAndGet();
                    }
                    Mode = ClientMode.Offline;
                    Logger.LogTrace("Message Handler done processing with status: COMMIT");
                }
                else if (IsFlushWaiting)
                {
                    Dropped.IncrementAndGet();
                    Logger.LogTrace("Message Handler done processing with status: BREAK");
                }
                else if (IsClosing)
                {
                    SetIsFailedWhileClosing(true);
                    Logger.LogTrace("Message Handler done processing with status: ABORT");
                }
                else
                {
                    Logger.LogTrace("Message Handler done processing with status: ABORT");
                }

            };
        int IAgent.DoWork()
        {
            int totalRead = 0;
            int batchRead;
            do
            {
                batchRead = _ringBuffer.Read(_ringBufferMessageHandler, BatchingSize);
                Producer.CompleteBatching();
                totalRead += batchRead;
            } while (batchRead > 0);
            if (totalRead > 0 && Logger.IsEnabled(LogLevel.Trace))
            {
                Logger.LogTrace("{TotalRead} messages consumed", totalRead);
            }
            return totalRead;
        }
        string IAgent.RoleName()
        {
            return "unity-buffer";
        }

    }
}
